package cn.metona.mq.consumer;

import cn.metona.mq.config.ConfigManager;
import cn.metona.mq.core.Message;
import cn.metona.mq.core.MessageQueue;
import cn.metona.mq.core.QueueMessage;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
import java.util.logging.Logger;

/**
 * 消息消费者
 * 支持自动消费、批量消费、并发消费等功能
 */
public class MessageConsumer {
    private static final Logger logger = Logger.getLogger(MessageConsumer.class.getName());

    private final String consumerId;                    // 消费者ID
    private final MessageQueue queue;                   // 关联的队列
    private final Consumer<Message> messageHandler;     // 消息处理器
    private final ExecutorService executorService;      // 执行线程池
    private final AtomicBoolean running;                // 运行状态
    private final int batchSize;                        // 批处理大小
    private final int concurrency;                      // 并发数
    private final ConfigManager configManager;          // 配置管理器
    private final boolean autoAck;                      // 是否自动确认

    // 用于控制暂停日志频率的计数器
    private final AtomicLong pauseLogCounter = new AtomicLong(0);
    private static final long PAUSE_LOG_INTERVAL = 100; // 每100次暂停中断打印一次日志

    /**
     * 构造函数
     */
    public MessageConsumer(String consumerId, MessageQueue queue,
                           Consumer<Message> messageHandler) {
        this(consumerId, queue, messageHandler, 1, 1, true);
    }

    /**
     * 构造函数 - 支持批处理和并发
     */
    public MessageConsumer(String consumerId, MessageQueue queue,
                           Consumer<Message> messageHandler,
                           int batchSize, int concurrency, boolean autoAck) {
        if (consumerId == null || consumerId.isEmpty()) {
            throw new IllegalArgumentException("Consumer ID cannot be null or empty");
        }
        if (queue == null) {
            throw new IllegalArgumentException("Queue cannot be null");
        }
        if (messageHandler == null) {
            throw new IllegalArgumentException("Message handler cannot be null");
        }
        if (batchSize <= 0) {
            throw new IllegalArgumentException("Batch size must be positive");
        }
        if (concurrency <= 0) {
            throw new IllegalArgumentException("Concurrency must be positive");
        }

        this.consumerId = consumerId;
        this.queue = queue;
        this.messageHandler = messageHandler;
        this.batchSize = batchSize;
        this.concurrency = concurrency;
        this.autoAck = autoAck;
        this.configManager = ConfigManager.getInstance();
        this.running = new AtomicBoolean(false);

        // 创建自定义线程工厂
        ThreadFactory threadFactory = new ThreadFactory() {
            private final AtomicInteger threadNumber = new AtomicInteger(1);

            @Override
            public Thread newThread(Runnable r) {
                Thread thread = new Thread(r, "Consumer-" + consumerId + "-" + threadNumber.getAndIncrement());
                thread.setDaemon(false);
                return thread;
            }
        };

        this.executorService = Executors.newFixedThreadPool(concurrency, threadFactory);

        // 将消费者添加到队列
        queue.addConsumer(this);
    }

    /**
     * 开始消费
     */
    public void start() {
        if (running.compareAndSet(false, true)) {
            logger.info("Consumer started: " + consumerId);
            pauseLogCounter.set(0); // 重置暂停日志计数器

            // 启动多个消费线程
            for (int i = 0; i < concurrency; i++) {
                executorService.submit(this::consumeLoop);
            }
        }
    }

    /**
     * 停止消费
     */
    public void stop() {
        if (running.compareAndSet(true, false)) {
            executorService.shutdown();
            queue.removeConsumer(this);
            logger.info("Consumer stopped: " + consumerId);
        }
    }

    /**
     * 消费循环
     */
    private void consumeLoop() {
        while (running.get() && !Thread.currentThread().isInterrupted()) {
            try {
                if (batchSize > 1) {
                    consumeBatch();
                } else {
                    consumeSingle();
                }
            } catch (MessageQueue.QueuePausedException e) {
                handleQueuePausedException(e);
            } catch (InterruptedException e) {
                handleInterruptedException(e);
            } catch (Exception e) {
                logger.severe("Error in consumer loop: " + e.getMessage());
                // 避免频繁错误导致CPU占用过高
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException ie) {
                    handleInterruptedException(ie);
                }
            }
        }
        logger.info("Consumer loop ended: " + consumerId);
    }

    /**
     * 处理中断异常
     */
    private void handleInterruptedException(InterruptedException e) {
        if (running.get()) {
            long count = pauseLogCounter.incrementAndGet();
            if (count % PAUSE_LOG_INTERVAL == 1) { // 每PAUSE_LOG_INTERVAL次打印一次
                logger.info("Consumer paused (interrupted), continuing: " + consumerId +
                        " (pause count: " + count + ")");
            }
        } else {
            logger.info("Consumer stopped: " + consumerId);
        }
        // 不要中断线程，继续循环
    }

    /**
     * 处理队列暂停异常
     */
    private void handleQueuePausedException(MessageQueue.QueuePausedException e) {
        long count = pauseLogCounter.incrementAndGet();
        if (count % PAUSE_LOG_INTERVAL == 1) { // 每PAUSE_LOG_INTERVAL次打印一次
            logger.info("Queue is paused, consumer waiting: " + consumerId +
                    " (pause count: " + count + ")");
        }
        // 等待一段时间再继续
        try {
            Thread.sleep(100);
        } catch (InterruptedException ie) {
            handleInterruptedException(ie);
        }
    }

    /**
     * 单条消费
     */
    private void consumeSingle() throws InterruptedException {
        QueueMessage queueMessage = queue.consumeMessage();
        if (queueMessage != null) {
            processMessage(queueMessage);
        }
    }

    /**
     * 批量消费
     */
    private void consumeBatch() throws InterruptedException {
        // 实现批量消费逻辑
        for (int i = 0; i < batchSize; i++) {
            QueueMessage queueMessage = queue.consumeMessageAsync();
            if (queueMessage != null) {
                processMessage(queueMessage);
            } else {
                break;
            }
        }
    }

    /**
     * 处理消息
     */
    private void processMessage(QueueMessage queueMessage) {
        if (queueMessage == null) {
            return;
        }

        try {
            Message message = queueMessage.getMessage();

            // 设置消费者ID
            queueMessage.setConsumerId(consumerId);

            // 检查消息是否过期
            if (message.isExpired()) {
                logger.warning("Message expired, discarding: " + message.messageId());
                if (autoAck) {
                    queue.ackMessage(message.messageId());
                }
                return;
            }

            // 处理消息
            messageHandler.accept(message);

            // 自动确认消息
            if (autoAck) {
                queue.ackMessage(message.messageId());
            }

            // 重置暂停日志计数器，因为成功处理了消息
            pauseLogCounter.set(0);

            logger.fine("Message processed successfully: " + message.messageId());

        } catch (Exception e) {
            logger.severe("Error processing message: " + queueMessage.getMessageId() +
                    ", error: " + e.getMessage());

            // 处理失败，根据重试次数决定是否重新入队
            if (autoAck) {
                queue.nackMessage(queueMessage.getMessageId());
            }
        }
    }

    /**
     * 手动确认消息
     */
    public boolean ackMessage(String messageId) {
        if (messageId == null || messageId.isEmpty()) {
            return false;
        }
        return queue.ackMessage(messageId);
    }

    /**
     * 手动拒绝消息
     */
    public boolean nackMessage(String messageId) {
        if (messageId == null || messageId.isEmpty()) {
            return false;
        }
        return queue.nackMessage(messageId);
    }

    /**
     * 获取消费者ID
     */
    public String getConsumerId() {
        return consumerId;
    }

    /**
     * 获取关联队列
     */
    public MessageQueue getQueue() {
        return queue;
    }

    /**
     * 是否正在运行
     */
    public boolean isRunning() {
        return running.get();
    }

    /**
     * 获取并发数
     */
    public int getConcurrency() {
        return concurrency;
    }

    /**
     * 获取批处理大小
     */
    public int getBatchSize() {
        return batchSize;
    }
}